package edu.sjtu.context.core;

import java.util.*;

import javax.jms.*;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
/**
 * Default topic manager. Use Apache ActiveMQ to create and maintain the context topic
 * @author Roger
 *
 */
public class DefaultTopicManager implements TopicManager {
	/**
	 * Topics under management.
	 */
	private Map<String, Topic> topics;
	
	/**
	 * Connection to a certain message broker. Please reference JMS API.
	 */
	private Connection connection;
	
	/**
	 * Connect session used to create topic and producer. Please reference JMS API.
	 */
	Session session;
	
	/**
	 * Message broker. Please reference JMS API.
	 */
	BrokerService broker;
	
	/**
	 * Static inner holder class, used to implement Singleton.
	 * @author Roger
	 *
	 */
	//static private class TopicManagerHolder{
		//private static volatile TopicManager tm = new DefaultTopicManager();
	//}
	
	/**
	 * Constructor. It will start the message broker.
	 */
	public DefaultTopicManager() {
		this.topics = new HashMap<String, Topic>();
	}
	
	/**
	 * Return the single instance.
	 * @return
	 */
	//public static TopicManager getInstance(){
		//return TopicManagerHolder.tm;
	//}
	
	/**
	 * Use to create a single topic. User can not use this method.
	 * If the topic is already exist, this method will not create the topic.
	 * @param topicName
	 * 		the name of the topic to be created.
	 * @exception
	 * 		if any problem occurred during the creation process, throw an exception.
	 */
	private void createTopic(String topicName)
	throws Exception{
		try{
			if(!this.topics.containsKey(topicName)){
				Topic topic = this.session.createTopic(topicName);
				this.topics.put(topicName, topic);
			}
		}catch(JMSException e){
			throw new Exception(e.getMessage());
		}
	}
	
	/**
	 * Use to create a set of topics. This method will be used by the user to create topics.
	 * @param topics
	 * 		name of topics to be created.
	 * @exception Exception
	 * 		if any problem occurred during the creation process, throw an exception.
	 */
	public void createTopics(String[] topics)
	throws Exception
	{
		for(String topic : topics)
			this.createTopic(topic);
	}
	
	/**
	 * Publish message to a certain topic, which is indicated by the parameter topicName. 
	 * If the topic is not exist, this method will create one, and then publish the message
	 * to the topic.
	 * @param topicName
	 * 		the name of the topic
	 * @param msg
	 * 		the message to be published
	 * @exception Exception
	 * 		if any problem occurred during the publish process, throw an exception.
	 */
	public void publish(String topicName, ContextMessage msg)
	throws Exception{
		try{
			if(!this.isExsit(topicName))
				this.createTopic(topicName);
			MessageProducer publisher = this.session.createProducer(this.topics.get(topicName));
			publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
			publisher.send(session.createTextMessage(msg.getMessage()));
		}catch(JMSException e){
			throw new Exception(e.getMessage());
		}
	}
	
	/**
	 * Use this method to get connection to a certain message broker.
	 * @param connectionUrl
	 * 		connection url to connect to message broker.
	 * @exception Exception
	 * 		if any problem occurred during create connection, throw an Exception
	 */
	public void getConnection(String connectionUrl)
	throws Exception{
		try{
			//initialize broker
			if(broker == null){
				broker = new BrokerService();
			}
			if(broker.isStarted())
				broker.stop();
			broker.addConnector(connectionUrl);
			broker.start();
			
			//initialize connection
			ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(connectionUrl);	
			this.connection = factory.createConnection();
			
			//initialize session
			this.session = this.connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		}catch(JMSException e){
			throw new Exception(e.getMessage());
		}
	}
	
	/**
	 * shot down the message broker.
	 */
	public void shutDown(){
		try{
		if(this.session != null)
			this.session.close();
		if(this.connection != null)
			this.connection.close();
		if(this.broker != null)
			this.broker.stop();
		}catch(Exception e){
			return;
		}
	}
	
	/**
	 * To check whether a certain topic is already exist.
	 * @param topicName
	 * 		the name of the topic.
	 * @return
	 * 		if the topic is already exist then return true, or return false;
	 */
	public boolean isExsit(String topicName){
		if(this.topics.containsKey(topicName)){
			return true;
		}else{
			return false;
		}
	}
}
